package com.cnebula.dataprocess.hive.job.consumer;

import org.apache.rocketmq.spring.starter.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.starter.core.RocketMQListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import com.cnebula.dataprocess.common.hive.HiveLoadData;
import com.cnebula.dataprocess.hive.job.serivce.HiveService;
import com.google.gson.Gson;

@Service
@RocketMQMessageListener(topic = "hive-loaddata-topic", consumerGroup = "hive-loaddata-consumer")
public class HiveLoadDataConsumer  implements RocketMQListener<HiveLoadData> {
	
	private static final Logger log = LoggerFactory.getLogger(HiveLoadDataConsumer.class);

	private Gson gson = new Gson();
	
	@Autowired
	HiveService hiveService;
	
	@Override
	public void onMessage(HiveLoadData arg0) {
		
		log.info("=============================receive hive loaddata job ============================");
		log.info(gson.toJson(arg0));
		log.info("===================================================================================");
		
		String datafile = arg0.getDatafile();
		String tableName = arg0.getHiveTable();
		int overwrite = arg0.getOverwrite();
		String jobId = arg0.getJobId();
		String taskId = arg0.getTaskId();
		if(overwrite == 1){
			hiveService.loadDataOverwrite(taskId, jobId, datafile, tableName);
		}else{
			hiveService.loadData(taskId, jobId, datafile, tableName);
		}
		
	}

}
